package com.niit.hjw.avgtotalandunitpricebyroom;

import com.niit.util.DataConvUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;


public class AvgTotalAndUnitPriceByRoomReducer extends Reducer<Text, Text, Text, Text> {
    private Text outv = new Text();
    private double totalPriceSum;
    private int unitPriceSum;
    private int total;// 相同户型的出现次数
    private List<String> list = new ArrayList<>();

    /**
     * reduce阶段的核心业务逻辑（根据相同户型 进行统计平均总价和单价，然后输出）
     *
     * @param key     户型, eg: 3 室 2 厅 2 卫
     * @param values  总价和单价, eg: 95, 13189元/㎡
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            System.out.println(value.toString());
            String[] datas = value.toString().split(",");
            totalPriceSum += DataConvUtils.strToDouble(datas[0].trim().split("\\[")[1], 0);
            unitPriceSum += DataConvUtils.strToInt(datas[1].trim().split("元")[0], 0);
            total++;
        }
        DecimalFormat df = new DecimalFormat("#.##");// 保留2位小数
        String avgTotalPrice = df.format(totalPriceSum / total);
        list.add(avgTotalPrice + "w");
        list.add((unitPriceSum / total) + "元/㎡");
        outv.set(list.toString());
        context.write(key, outv);

        list.clear();
        total = 0;
        unitPriceSum = 0;
        totalPriceSum = 0;
    }
}
